package cn.tecnova.cd

import cn.tecnova.utils.ConfigHandler
import org.apache.log4j.{Level, Logger}
import org.apache.spark.{SparkConf, SparkContext}
import org.apache.spark.sql.{Dataset, SQLContext}
import org.elasticsearch.spark.rdd.EsSpark

/**
  * description:
  * 6812 9820
  **/
object Base2Es {

  Logger.getLogger("org").setLevel(Level.ERROR)

  def main(args: Array[String]): Unit = {

    val conf = new SparkConf()
      .setAppName(this.getClass.getSimpleName)
      //      .setMaster("local[*]")
      .set("es.index.auto.create", "true")
      .set("es.nodes", ConfigHandler.esNodes)
      .set("es.port", ConfigHandler.esPort)
      .set("es.nodes.wan.only", "true")

    val sc = new SparkContext(conf)
    val sqLContext = new SQLContext(sc)

    import sqLContext.implicits._
    val rowDF = sqLContext.read.jdbc("jdbc:mysql://www.slave4.com:3306/tec_base2", "base","id",1,68129820,10, ConfigHandler.props2)

    val result: Dataset[Base] = rowDF.map(row => {
      Base(
        row.getAs[Int]("id"),
        row.getAs[String]("keyNo"),
        row.getAs[String]("name"),
        row.getAs[String]("no"),
        row.getAs[String]("belongOrg"),
        row.getAs[String]("operName"),
        row.getAs[String]("startDate"),
        row.getAs[String]("endDate"),
        row.getAs[String]("status"),
        row.getAs[String]("updatedDate"),
        row.getAs[String]("shortStatus"),
        row.getAs[String]("creditCode"),
        row.getAs[String]("registCapi"),
        row.getAs[String]("econKind"),
        row.getAs[String]("industryCode"),
        row.getAs[String]("subIndustryCode"),
        row.getAs[String]("industry"),
        row.getAs[String]("subIndustry"),
        row.getAs[String]("scope"),
        row.getAs[String]("contactNumber"),
        row.getAs[String]("email"),
        row.getAs[String]("webSite"),
        row.getAs[String]("imageUrl"),
        row.getAs[String]("orgNo"),
        row.getAs[String]("englishName"),
        row.getAs[String]("product"),
        row.getAs[String]("type"),
        row.getAs[String]("originalName"),
        row.getAs[String]("hitField"),
        row.getAs[String]("hitValue"),
        row.getAs[String]("address"),
        row.getAs[String]("provinceCode"),
        row.getAs[String]("cityCode"),
        row.getAs[String]("countyCode"),
        row.getAs[String]("province"),
        row.getAs[String]("city"),
        row.getAs[String]("county"),
        row.getAs[String]("registCapiNum")
      )
    })

    EsSpark.saveToEs(result.rdd,"tecnova_base/base")

    sc.stop()

  }


}
